package net.bw.realtime.warehouse.dwd;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bw.realtime.warehouse.dwd
 * @Author: zhangleqing
 * @CreateTime: 2025-08-26  11:23
 * @Description: TODO dwd_summary_entrance_guard_log （门禁类型汇总表）
 * @Version: 1.0
 */
public class DwdEntranceGuardLog {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 设置检查点
//        env.enableCheckpointing(1000);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);


        // todo 1 门禁刷卡事实表 ods_entrance_guard_log_clean
        tEnv.executeSql(
                "CREATE TABLE ods_entrance_guard_log_clean(\n" +
                        " id STRING NOT NULL,\n" +
                        " openMode STRING,\n" +
                        " openTime TIMESTAMP(3),\n" +
                        " ownerId INT,\n" +
                        " proc_time as PROCTIME(),\n" +
                        " WATERMARK FOR openTime AS openTime - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        " 'connector' = 'kafka',\n" +
                        " 'topic' = 'ods_entrance_guard_log_clean',\n" +
                        " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                        " 'properties.group.id' = 'ods_entrance_guard_log_clean',\n" +
                        " 'scan.startup.mode' = 'earliest-offset',\n" +
                        " 'format' = 'json'\n" +
                        ")"
        );

        // todo 2 用户维表 ods_ower_clean
        tEnv.executeSql(
                "CREATE TABLE ods_ower_clean(\n" +
                        " id BIGINT NOT NULL,\n" +
                        " roomId BIGINT NOT NULL,\n" +
                        " owerCode STRING,\n" +
                        " owerName STRING,\n" +
                        " owerSex STRING,\n" +
                        " owerAge INT,\n" +
                        " owerIdcard STRING,\n" +
                        " owerTel BIGINT,\n" +
                        " owerHomeplace STRING,\n" +
                        " owerWorkplace STRING,\n" +
                        " remark STRING,\n" +
                        " createTime TIMESTAMP(3),\n" +
                        " operatorTime TIMESTAMP(3),\n" +
                        " proc_time as PROCTIME(),\n" +
                        " WATERMARK FOR createTime AS createTime - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        " 'connector' = 'kafka',\n" +
                        " 'topic' = 'ods_ower_clean',\n" +
                        " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                        " 'properties.group.id' = 'ods_ower_clean',\n" +
                        " 'scan.startup.mode' = 'earliest-offset',\n" +
                        " 'format' = 'json'\n" +
                        ")"
        );

        // todo 3 房间维表 ods_room_clean
        tEnv.executeSql(
                "CREATE TABLE ods_room_clean(\n" +
                        " id BIGINT NOT NULL,\n" +
                        " houseId BIGINT NOT NULL,\n" +
                        " roomCode STRING,\n" +
                        " roomName STRING,\n" +
                        " peopleNum BIGINT,\n" +
                        " remark STRING,\n" +
                        " proc_time as PROCTIME()\n" +
                        ") WITH (\n" +
                        " 'connector' = 'kafka',\n" +
                        " 'topic' = 'ods_room_clean',\n" +
                        " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                        " 'properties.group.id' = 'ods_room_clean',\n" +
                        " 'scan.startup.mode' = 'earliest-offset',\n" +
                        " 'format' = 'json'\n" +
                        ")"
        );

// ==========================================《维度表》============================================

        // todo 4 楼栋维表 house
        tEnv.executeSql(
                "CREATE TABLE ods_house (\n" +
                        "    id INT,\n" +
                        "    cid INT,\n" +
                        "    houseCode STRING,\n" +
                        "    houseName STRING,\n" +
                        "    buildStartTime STRING,\n" +
                        "    buildEndTime STRING,\n" +
                        "    houseArea STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_two.house',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );

        // todo 5 社区维表 community
        tEnv.executeSql(
                "CREATE TABLE ods_community (\n" +
                        "    id INT,\n" +
                        "    communityName STRING,\n" +
                        "    district_id INT,\n" +
                        "    city_id INT,\n" +
                        "    communityLocation STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_two.community',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );

        // todo 6 地区维表 district
        tEnv.executeSql(
                "CREATE TABLE ods_district (\n" +
                        "    id INT,\n" +
                        "    pid INT,\n" +
                        "    code STRING,\n" +
                        "    name STRING,\n" +
                        "    remark STRING,\n" +
                        "    create_time STRING,\n" +
                        "    update_time STRING,\n" +
                        "    status STRING,\n" +
                        "    level STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_two.district',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );


/*
COALESCE
SELECT
    log.id,
    coalesce(dist.name, '未知') as districtName,
    coalesce(comm.communityName, '未知') as communityName,
    coalesce(house.houseCode, '未知') as houseCode,
    coalesce(room.roomCode, '未知') as roomCode,
    coalesce(ower.id, -1) as ownerId,
    log.openMode as openMode,
    log.openTime
FROM ods_entrance_guard_log_clean log
-- 关联用户维表
LEFT JOIN ods_ower_clean ower
ON log.ownerId = ower.id
-- 关联房间维表
LEFT JOIN ods_room_clean room
ON ower.roomId = room.id
-- 关联楼栋维表
LEFT JOIN ods_house house
ON room.houseId = house.id
-- 关联社区维表，获取社区ID(cId)
LEFT JOIN ods_community comm
ON house.cid = comm.id
-- 获取地区维表
left join ods_district dist
on comm.district_id = dist.id
 */
        // todo dwd_summary_entrance_guard_log  门禁类型汇总表
        Table dwd_summary_entrance_guard_log = tEnv.sqlQuery(
                "SELECT\n" +
                        "    log.id,\n" +
                        "    COALESCE(dist.name, '未知') as districtName,\n" +
                        "    COALESCE(comm.communityName, '未知') as communityName,\n" +
                        "    COALESCE(house.houseCode, '未知') as houseCode,\n" +
                        "    COALESCE(room.roomCode, '未知') as roomCode,\n" +
                        "    COALESCE(ower.id, -1) as ownerId,\n" +
                        "    log.openMode as openMode,\n" +
                        "    log.openTime\n" +
                        "FROM ods_entrance_guard_log_clean log\n" +
                        "-- 关联用户维表\n" +
                        "LEFT JOIN ods_ower_clean ower\n" +
                        "ON log.ownerId = ower.id\n" +
                        "-- 关联房间维表\n" +
                        "LEFT JOIN ods_room_clean room\n" +
                        "ON ower.roomId = room.id\n" +
                        "-- 关联楼栋维表\n" +
                        "LEFT JOIN ods_house house\n" +
                        "ON room.houseId = house.id\n" +
                        "-- 关联社区维表，获取社区ID(cId)\n" +
                        "LEFT JOIN ods_community comm\n" +
                        "ON house.cid = comm.id\n" +
                        "-- 获取地区维表\n" +
                        "left join ods_district dist\n" +
                        "on comm.district_id = dist.id"
        );

        // 创建临时视图
        tEnv.createTemporaryView("dwd_summary_entrance_guard_log", dwd_summary_entrance_guard_log);

        // 打印前 10 条数据
        tEnv.sqlQuery("select * from dwd_summary_entrance_guard_log limit 10")
                        .execute().print();


        env.execute("DwdEntranceGuardLog");
    }
}
